package com.example.hystrixDemo.rxjava;/**
 * @Deacription TODO
 * @Author baohuajie
 * @Date 2020/12/14 22:55
 * @Version 1.0
 **/

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;
import io.reactivex.schedulers.Schedulers;

/**
 * @Deacription TODO
 * @Author baohuajie
 * @Date 2020/12/14 22:55
 * @Version 1.0
 **/
public class Main1 {

    public static void main(String[] args) {

        /**
         * 注：整体方法调用顺序：观察者.onSubscribe（）> 被观察者.subscribe（）> 观察者.onNext（）>观察者.onComplete()
         */
        /**
         * 线程切换需要注意的
         RxJava 内置的线程调度器的确可以让我们的线程切换得心应手，但其中也有些需要注意的地方。

         简单地说，subscribeOn() 指定的就是发射事件的线程，observerOn 指定的就是订阅者接收事件的线程。
         多次指定发射事件的线程只有第一次指定的有效，也就是说多次调用 subscribeOn() 只有第一次的有效，其余的会被忽略。
         但多次指定订阅者接收线程是可以的，也就是说每调用一次 observerOn()，下游的线程就会切换一次。
         */

        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                for (int i = 0; i < 1000; i++) {
                    System.out.println("我是小说，我更新了第" + i + "季" + "线程" + Thread.currentThread().getName());
                    emitter.onNext(i + "");
                }
            }
        }).observeOn(Schedulers.io())
                .subscribe(new Observer<String>() {

                    private Disposable disposable = null;

                    @Override
                    public void onSubscribe(Disposable d) {
                        disposable = d;
                        System.out.println("我是读者，我和小说订阅了");
                    }

                    @Override
                    public void onNext(String s) {
                        /*if (Integer.valueOf(s) == 10) {
                            System.out.println("我是读者，我解除了新版本：" + s + "版本" + "线程" + Thread.currentThread().getName());
                            disposable.dispose();
                        }*/
                        System.out.println("我是读者，我拿到了小说的新版本：" + s + "版本" + "线程" + Thread.currentThread().getName());
                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {
                        System.out.println("我是读者，小说的新版本被我拿完了");
                    }
                });
    }
}
